package com.sunday.common.mq.kafka.study.spring.e9_Producer_Consumer_Interceptors;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.DefaultKafkaConsumerFactoryCustomizer;
import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.kafka.support.converter.RecordMessageConverter;

import java.util.Map;

/**
 * https://docs.spring.io/spring-kafka/docs/current/reference/html/#interceptors
 */
@Slf4j
@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Bean
    public NewTopic topic6() {
        return TopicBuilder.name("topic6")
                .partitions(3)
                .replicas(1)
                .build();
    }

    @Bean
    @ConditionalOnMissingBean(ConsumerFactory.class)
    public DefaultKafkaConsumerFactory<?, ?> kafkaConsumerFactory(KafkaConnectionDetails connectionDetails,
                                                                  ObjectProvider<DefaultKafkaConsumerFactoryCustomizer> customizers,
                                                                  KafkaProperties prop
    ) {
        Map<String, Object> properties = prop.buildConsumerProperties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getConsumerBootstrapServers());

        properties.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MyConsumerInterceptor.class.getName());
        properties.put(SomeBean.NAME, SomeBean.INSTANCE);

        DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>(properties);
        customizers.orderedStream().forEach((customizer) -> customizer.customize(factory));
        return factory;
    }


    /**
     * 方法 一. 将Spring Bean连接到生产者/消费者拦截器
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#interceptors
     */
//    @Bean
//    @ConditionalOnMissingBean(ProducerFactory.class)
//    public DefaultKafkaProducerFactory<?, ?> kafkaProducerFactory(KafkaConnectionDetails connectionDetails,
//                                                                  ObjectProvider<DefaultKafkaProducerFactoryCustomizer> customizers,
//                                                                  KafkaProperties prop
//    ) {
//        Map<String, Object> properties = prop.buildProducerProperties();
//        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getProducerBootstrapServers());
//
//        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName());
//        properties.put(SomeBean.NAME, SomeBean.INSTANCE);
//
//        DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(properties);
//        String transactionIdPrefix = prop.getProducer().getTransactionIdPrefix();
//        if (transactionIdPrefix != null) {
//            factory.setTransactionIdPrefix(transactionIdPrefix);
//        }
//        customizers.orderedStream().forEach((customizer) -> customizer.customize(factory));
//        return factory;
//    }

    /**
     * 方法 二. Spring 中管理的 Producer Interceptor
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#producer-interceptor-managed-in-spring
     */
    @Bean
    public MyProducerInterceptor myProducerInterceptor() {
        MyProducerInterceptor producerInterceptor = new MyProducerInterceptor();
        producerInterceptor.configure(Map.of(SomeBean.NAME, SomeBean.INSTANCE));
        return producerInterceptor;
    }

    @Bean
    @ConditionalOnMissingBean(KafkaTemplate.class)
    public KafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> kafkaProducerFactory,
                                             ProducerListener<Object, Object> kafkaProducerListener,
                                             ObjectProvider<RecordMessageConverter> messageConverter,
                                             KafkaProperties prop,
                                             MyProducerInterceptor myProducerInterceptor
    ) {
        PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
        KafkaTemplate<Object, Object> kafkaTemplate = new KafkaTemplate<>(kafkaProducerFactory);
        messageConverter.ifUnique(kafkaTemplate::setMessageConverter);
        map.from(kafkaProducerListener).to(kafkaTemplate::setProducerListener);
        map.from(prop.getTemplate().getDefaultTopic()).to(kafkaTemplate::setDefaultTopic);
        map.from(prop.getTemplate().getTransactionIdPrefix()).to(kafkaTemplate::setTransactionIdPrefix);

        kafkaTemplate.setProducerInterceptor(myProducerInterceptor);

        return kafkaTemplate;
    }

}
